package com.wry.listener;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.wry.entity.WOrder;
import com.wry.service.OrderProducerService;
import com.wry.service.WOrderService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <h1> 订单服务的事务消息监听器 <h1>
 *
 * @author 魏如元
 * @since 2022/10/14
 */
@Component
@Slf4j
public class TransactionProducer implements InitializingBean {

    @Resource
    private WOrderService wOrderService;

    @Resource
    private OrderProducerService orderProducerService;

    @Value("${rocketmq.name-server}")
    private String addr;

    @Value("${rocketmq.producer.group}")
    private String group;

    public static final TransactionMQProducer producer = new TransactionMQProducer("transactional-order-group-2");

    @Override
    public void afterPropertiesSet() throws MQClientException {
//        TransactionMQProducer producer = new TransactionMQProducer(group);
        producer.setNamesrvAddr(addr);

        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("transaction-thread-name-%s").build();
        // 构建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 60,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(30), threadFactory);
        producer.setExecutorService(executor);

        // 设置发送消息的回调
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 根据消息发送的结果 判断是否执行本地事务
             *
             * 回调该方法的时候说明 消息已经成功发送到了MQ，可以把订单状态更新为 "支付成功"
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                // 根据本地事务执行成与否判断 事务消息是否需要commit与 rollback
                ObjectMapper objectMapper = new ObjectMapper();
                LocalTransactionState state;
                try {
                    OrderRecord record = objectMapper.readValue(message.getBody(), OrderRecord.class);
                    //MQ已经收到了TransactionProducer send方法发送的事务消息，下面执行本地的事务
                    //本地记录订单信息
                    orderProducerService.updatePayStatusByOrderId(record.getOrderId());

                    // 提交事务
                    state = LocalTransactionState.COMMIT_MESSAGE;
                }catch (Exception e){
                    e.printStackTrace();
                    // 回滚事务
                    state = LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return state;
            }

            /**
             * RocketMQ 回调 根据本地事务是否执行成功 告诉broker 此消息是否投递成功
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                ObjectMapper objectMapper = new ObjectMapper();
                LocalTransactionState state;
                try {
                    OrderRecord record = objectMapper.readValue(messageExt.getBody(), OrderRecord.class);
                    //根据是否有transaction_id对应转账记录 来判断事务是否执行成功
                    boolean isLocalSuccess = orderProducerService.checkOrderPaySuccess(record.getOrderId());
                    log.info("orderId = 【{}】，the payStatus is {}", record.getOrderId(), isLocalSuccess);
                    state = isLocalSuccess ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE;
                }catch (Exception e){
                    state = LocalTransactionState.ROLLBACK_MESSAGE;
                    e.printStackTrace();
                }
                return state;
            }
        });
    }

    /**
     * 发送支付成功事务消息
     */
    public void sendOrderPaySuccessEvent(String orderId) throws JsonProcessingException, UnsupportedEncodingException, MQClientException {
        producer.setNamesrvAddr(addr);
        final WOrder order = wOrderService.lambdaQuery()
                .eq(WOrder::getOrderId, orderId)
                .one();
        if (Objects.isNull(order)){
            log.warn("orderId = {}，not found order",orderId);
            return;
        }
        // 构造发送的事务 消息
        OrderRecord record = new OrderRecord();
        record.setUserId(order.getUserId());
        record.setOrderId(orderId);
        record.setBuyNum(order.getBuyNum());
        record.setPayStatus(order.getPayStatus());
        record.setGoodId(order.getGoodId());
        producer.start();
        Message message = new Message("Order-Success", "", record.getOrderId(),
                new ObjectMapper().writeValueAsString(record).getBytes(RemotingHelper.DEFAULT_CHARSET));
        TransactionSendResult result = producer.sendMessageInTransaction(message, null);

        log.info("发送事务消息 , orderId = {} , result = {}",record.getOrderId(), result);
    }

    @Data
    public static class OrderRecord{
        private String orderId;

        /**
         * 购买数量
         */
        private Integer buyNum;

        /**
         * 商品ID
         */
        private Integer goodId;

        /**
         * 用户ID
         */
        private Integer userId;

        /**
         * 支付状态，0：没有支付，1：已经支付
         */
        private Integer payStatus;
    }
}
